package com.sdp.irrigation.kafka.biz;

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.sdp.core.bizc.redis.RedisUtil;
import com.sdp.irrigation.db.irrigation_eq_command_downlink.service.IIrrigationEqCommandDownlinkService;
import com.sdp.irrigation.entity.IrrigationEqCommandDownlink;
import com.sdp.irrigation.kafka.producer.KafkaProducer;
import com.sdp.irrigation.kafka.property.ServerProperty;
import com.sdp.irrigation.kafka.task.EqStatusQueryTask;
import com.sdp.irrigation.kafka.utils.ProtocolUtils;
import com.sdp.irrigation.protocol.CommPackage;
import com.sdp.irrigation.timer.NettyTimer;
import com.sdp.irrigation.utils.DownlinkCodeEnum;
import com.sdp.irrigation.utils.JsonUtils;
import com.sdp.irrigation.view.DownAgitationParam;
import com.sdp.irrigation.view.DownDesiltingParam;
import com.sdp.irrigation.view.DownEqCloseParam;
import com.sdp.irrigation.view.DownEqOpenParam;
import com.sdp.irrigation.view.DownFertigationParam;
import com.sdp.irrigation.view.DownSettleParam;
import com.sdp.irrigation.view.DownValveParam;
import com.sdp.irrigation.view.DownWaterSupplyParam;

@Component
public class CommandDownlinkBiz {

	@Autowired
	private IIrrigationEqCommandDownlinkService eqCommandDownlinkService;
	
	@Autowired
	private RedisUtil redisUtil;
	
	@Autowired
	private ServerProperty serverProperty;
	
	/**
	 * 下发打开水泵
	 * @param downEqOpenParam
	 * @param kafkaProducer
	 */
	public void downlinkCommandEqOpen(DownEqOpenParam downEqOpenParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownEqOpenCommPackage(downEqOpenParam);
		String msg = JsonUtils.toJSONString(commPackage);
		
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downEqOpenParam.getEqNo(), downEqOpenParam.getIcNo(), downEqOpenParam.getDownlinkType(), DownlinkCodeEnum.PUMP_ON, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downEqOpenParam.getEqNo(), msg, kafkaProducer);	
	}
	
	/**
	 * 下发关闭水泵
	 * @param downEqCloseParam
	 * @param kafkaProducer
	 */
	public void downlinkCommandEqClose(DownEqCloseParam downEqCloseParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownEqCloseCommPackage(downEqCloseParam);
		String msg = JsonUtils.toJSONString(commPackage);
		
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downEqCloseParam.getEqNo(), downEqCloseParam.getIcNo(), downEqCloseParam.getDownlinkType(), DownlinkCodeEnum.PUMP_OFF, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downEqCloseParam.getEqNo(), msg, kafkaProducer);	
	}
	
	private IrrigationEqCommandDownlink initCommandDownlinkRecord(String eqNo, String icNo, String downlinkType, DownlinkCodeEnum downlinkCodeEnum, String msg) {
		IrrigationEqCommandDownlink eqCommandDownlink = new IrrigationEqCommandDownlink();
		eqCommandDownlink.setEqNo(eqNo);
		eqCommandDownlink.setIcNo(icNo);
		eqCommandDownlink.setDownlinkType(downlinkType);
		eqCommandDownlink.setCommandNo(String.valueOf(downlinkCodeEnum.getCode()));
		eqCommandDownlink.setCommandName(downlinkCodeEnum.getName());
		eqCommandDownlink.setCommandContent(msg);
		eqCommandDownlink.setDownlinkTime(LocalDateTime.now());
		eqCommandDownlink.setCreateTime(eqCommandDownlink.getDownlinkTime());
		eqCommandDownlink.setUpdateTime(eqCommandDownlink.getDownlinkTime());
		
		return eqCommandDownlink;
	}
	
	private void sendAndOpenQueryTask(String commandNo, String eqNo, String msg, KafkaProducer kafkaProducer) {
		//发送kafka
		kafkaProducer.send(ProtocolUtils.getDownMessageTopic(commandNo, serverProperty.getDownTopicPrefix()), msg);
				
		EqStatusQueryTask eqStatusQueryTask = new EqStatusQueryTask(redisUtil, eqNo, kafkaProducer, serverProperty);
		if (eqStatusQueryTask.isOpenTask(eqNo)) {
			NettyTimer.newTimeout(eqStatusQueryTask, EqStatusQueryTask.QUERY_DELAY, TimeUnit.SECONDS);
		}
	}
	
	/**
	 * 下发开关阀门
	 * @param downValveParam
	 * @param kafkaProducer
	 */
	public void downlinkCommandValve(DownValveParam downValveParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownValveCommPackage(downValveParam);
		String msg = JsonUtils.toJSONString(commPackage);
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downValveParam.getEqNo(), downValveParam.getIcNo(), downValveParam.getDownlinkType(), DownlinkCodeEnum.VALVE_ON_OFF, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downValveParam.getEqNo(), msg, kafkaProducer);
		
	}
	
	public void downlinkCommandWaterSupply(DownWaterSupplyParam downWaterSupplyParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownWaterSupplyCommPackage(downWaterSupplyParam);
		String msg = JsonUtils.toJSONString(commPackage);
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downWaterSupplyParam.getEqNo(), downWaterSupplyParam.getIcNo(), downWaterSupplyParam.getDownlinkType(), DownlinkCodeEnum.WATER_SUPPLY, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downWaterSupplyParam.getEqNo(), msg, kafkaProducer);
		
	}
	
	public void downlinkCommandAgitation(DownAgitationParam downAgitationParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownAgitationCommPackage(downAgitationParam);
		String msg = JsonUtils.toJSONString(commPackage);
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downAgitationParam.getEqNo(), downAgitationParam.getIcNo(), downAgitationParam.getDownlinkType(), DownlinkCodeEnum.AGITATION, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downAgitationParam.getEqNo(), msg, kafkaProducer);
		
	}
	
	public void downlinkCommandFertigation(DownFertigationParam downFertigationParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownFertigationCommPackage(downFertigationParam);
		String msg = JsonUtils.toJSONString(commPackage);
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downFertigationParam.getEqNo(), downFertigationParam.getIcNo(), downFertigationParam.getDownlinkType(), DownlinkCodeEnum.FERTIGATION, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downFertigationParam.getEqNo(), msg, kafkaProducer);
		
	}
	
	public void downlinkCommandDesilting(DownDesiltingParam downDesiltingParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownDesiltingCommPackage(downDesiltingParam);
		String msg = JsonUtils.toJSONString(commPackage);
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downDesiltingParam.getEqNo(), downDesiltingParam.getIcNo(), downDesiltingParam.getDownlinkType(), DownlinkCodeEnum.DESILTING, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		
		sendAndOpenQueryTask(eqCommandDownlink.getCommandNo(), downDesiltingParam.getEqNo(), msg, kafkaProducer);
		
	}
	
	public void downlinkCommandSettle(DownSettleParam downSettleParam, KafkaProducer kafkaProducer) {
		CommPackage commPackage = ProtocolUtils.getDownSettleCommPackage(downSettleParam);
		String msg = JsonUtils.toJSONString(commPackage);
		
		IrrigationEqCommandDownlink eqCommandDownlink = initCommandDownlinkRecord(downSettleParam.getEqNo(), downSettleParam.getIcNo(), downSettleParam.getDownlinkType(), DownlinkCodeEnum.SETTLE, msg);
		eqCommandDownlinkService.save(eqCommandDownlink);
		//暂时不在下发report报文，这里设备应该会上报user报文
		//发送kafka
		kafkaProducer.send(ProtocolUtils.getDownMessageTopic(eqCommandDownlink.getCommandNo(), serverProperty.getDownTopicPrefix()), msg);
	}
	
}
